package com.atguigu.fail.detect;

import com.atguigu.fail.detect.bean.LoginEvent;
import com.atguigu.fail.detect.bean.LoginFailWarning;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;

public class LoginFailWithCep {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        //1.从文件中读取数据
        URL resource = LoginFailWithCep.class.getResource("/LoginLog.csv");
        DataStreamSource<String> dataStreamSource = env.readTextFile(resource.getPath());
        SingleOutputStreamOperator<LoginEvent> loginEventStream = dataStreamSource.map(line -> {
            String[] fields = line.split(",");
            return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
            @Override
            public long extractTimestamp(LoginEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        //自定义处理函数，检测连续登陆失败事件
        /*SingleOutputStreamOperator<LoginFailWarning> warningStream = loginEventStream
                .keyBy(LoginEvent::getUserId)
                .process(new LoginFail.LoginFailDetectWarning(2));*/

        //1.定义个匹配模式
        // first fail -> second fail,within 2 seconds
        Pattern<LoginEvent, LoginEvent> LoginFailPattern0 = Pattern
                .<LoginEvent>begin("firstFail").where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return "fail".equals(loginEvent.getLoginState());
                    }
                }).next("secondFail").where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return "fail".equals(loginEvent.getLoginState());
                    }
                }).next("thirdFail").where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return "fail".equals(loginEvent.getLoginState());
                    }
                }).within(Time.seconds(3));

        Pattern<LoginEvent, LoginEvent> LoginFailPattern = Pattern
                .<LoginEvent>begin("failEvents").where(
                        new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return "fail".equals(loginEvent.getLoginState());
                    }
                }).times(3)
                //连续3次
                .consecutive()
                .within(Time.seconds(5));
        //2.将匹配模式应用到数据流上，得到一个pattern stream
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventStream.keyBy(LoginEvent::getUserId), LoginFailPattern);

        //3.从pattern stream中检出符合条件的复杂事件，进行转换处理
        SingleOutputStreamOperator<LoginFailWarning> warningStream = patternStream.select(new LoginFailPatternMatchDetect());

        warningStream.print();
        env.execute("login event warning with cep");
    }

    public static class LoginFailPatternMatchDetect implements PatternSelectFunction<LoginEvent, LoginFailWarning> {

        @Override
        public LoginFailWarning select(Map<String, List<LoginEvent>> map) throws Exception {
            /*LoginEvent firstFailEvent = map.get("firstFail").get(0);
            LoginEvent lastFailEvent = map.get("thirdFail").iterator().next();

            return new LoginFailWarning(firstFailEvent.getUserId()
                    ,firstFailEvent.getTimestamp()
                    ,lastFailEvent.getTimestamp()
            ,"login failed "+map.size()+" times in 3 seconds");*/

            LoginEvent firstFailEvent = map.get("failEvents").get(0);
            LoginEvent lastFailEvent = map.get("failEvents").get(map.get("failEvents").size() - 1);
            return new LoginFailWarning(firstFailEvent.getUserId()
                    , firstFailEvent.getTimestamp()
                    , lastFailEvent.getTimestamp()
                    , "login failed " + map.get("failEvents").size() + " times in 3 seconds");
        }
    }
}
